package com.gzyj.flink.gps;

import cn.hutool.core.collection.CollectionUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.parser.ParserConfig;
import com.alibaba.fastjson.parser.deserializer.ObjectDeserializer;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.alibaba.fastjson.support.spring.FastJsonRedisSerializer;
import com.gzyj.flink.broadcast.DaoludengjiBroadcastFun;
import com.gzyj.flink.driver.Driverlogininfo;
import com.gzyj.flink.gp.zdaq.TCmd020064;
import com.gzyj.flink.gp.zdaq.TCmd020065;
import com.gzyj.flink.gps.addtionalfnfo.AbstractResultDeserializer;
import com.gzyj.flink.gps.addtionalfnfo.JTT808_0x0200_0x30;
import com.gzyj.flink.gps.addtionalfnfo.JTT808_0x0200_0x31;
import com.gzyj.flink.gps.addtionalfnfo.JTT808_AddtionalInfo;
import com.gzyj.flink.gps.broadcast.TCmd020064Process;
import com.gzyj.flink.gps.broadcast.TCmd020065Process;
import com.gzyj.flink.gps.config.RedisKeyPrefixConfig;
import com.gzyj.flink.gps.factory.RedisSessionFactory;
import com.gzyj.flink.gps.map.*;
import com.gzyj.flink.gps.sink.gp.BaobiaoAlarmsummaryRecordSink;
import com.gzyj.flink.gps.sink.gp.MybatisGPSSink;
import com.gzyj.flink.gps.sink.gp.TCmd020064Sink;
import com.gzyj.flink.gps.sink.gp.TCmd020065Sink;
import com.gzyj.flink.pojo.gps.*;
import com.gzyj.flink.source.mysql.DriverLoginInfoSource;
import com.gzyj.flink.source.mysql.MonDaoludengjiSetSource;
import com.gzyj.flink.utils.GpsIntUtils;
import com.gzyj.flink.utils.PointUtils;
import com.gzyj.flink.utils.RoadUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.CoGroupFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.IterativeCondition;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;

import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase;
import org.apache.flink.util.Collector;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.locationtech.jts.geom.Coordinate;
import org.locationtech.jts.geom.GeometryFactory;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;

import java.lang.reflect.Type;
import java.math.BigDecimal;
import java.util.*;


/**
 * @author lhj
 */
public class GPSApp {
    public static void main(String[] args) throws Exception {
        Configuration configuration = new Configuration();

        configuration.setString("mysql-url", "jdbc:mysql://127.0.0.1:3306/blade?useServerPrepStmts=false&rewriteBatchedStatements=true&useUnicode=true&serverTimezone=GMT%2B8&characterEncoding=utf-8&autoReconnect=true&failOverReadOnly=false&useCompression=true");
        configuration.setString("mysql-user", "root");
        configuration.setString("mysql-password", "123456");

        configuration.setString("tidb-url", "jdbc:mysql://127.0.0.1:4000/gzyj_daoluyunshu?useServerPrepStmts=false&rewriteBatchedStatements=true&useUnicode=true&serverTimezone=GMT%2B8&characterEncoding=utf-8&autoReconnect=true&failOverReadOnly=false&useCompression=true");
        configuration.setString("tidb-user", "root");
        configuration.setString("tidb-password", "123456");

        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        executionEnvironment.setParallelism(10);

        executionEnvironment.getConfig().setGlobalJobParameters(configuration);

        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "127.0.0.1:9092");
        properties.setProperty("group.id", "flink");
        properties.setProperty("key.deserializer", StringDeserializer.class.getName());
        properties.setProperty("value.deserializer", StringDeserializer.class.getName());

        FlinkKafkaConsumerBase<String> kafkasource = new FlinkKafkaConsumer<String>("flink-gps", new SimpleStringSchema(), properties);

        DataStream<String> gpsStream = executionEnvironment.addSource(kafkasource);

        DataStream<GPSPosition> gpsPositionDataStream = gpsStream.map(new MapFunction<String, GPSPosition>() {
            @Override
            public GPSPosition map(String s) throws Exception {
                JSONObject jsonObject = JSON.parseObject(s);

                ParserConfig parserConfig = new ParserConfig() {
                    @Override
                    public ObjectDeserializer getDeserializer(Type type) {
                        if (type == JTT808_AddtionalInfo.class) {
                            return new AbstractResultDeserializer("2013");
                        }
                        return super.getDeserializer(type);
                    }
                };

                if (jsonObject.containsKey("version")
                        &&
                        jsonObject.getString("version").equalsIgnoreCase("2019")) {
                    parserConfig = new ParserConfig() {
                        @Override
                        public ObjectDeserializer getDeserializer(Type type) {
                            if (type == JTT808_AddtionalInfo.class) {
                                return new AbstractResultDeserializer("2019");
                            }
                            return super.getDeserializer(type);
                        }
                    };

                }

                GPSPosition gpsPosition = JSON.parseObject(s, GPSPosition.class, parserConfig);

                return gpsPosition;
            }
        });


        SingleOutputStreamOperator<GPSPosition> gpsSingleStream = gpsPositionDataStream.map(new IsInDatabase()).filter(Objects::nonNull);


        SingleOutputStreamOperator<VehGps> gpsSingleOutputStreamOperator = gpsSingleStream.map(new RichMapFunction<GPSPosition, VehGps>() {
            LettuceConnectionFactory redisFactory;
            private RedisTemplate<String, Object> getRedisTemplate(LettuceConnectionFactory redisSessionFactory) {
                RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
                redisTemplate.setConnectionFactory(redisSessionFactory);
                redisTemplate.setKeySerializer(new StringRedisSerializer());
                redisTemplate.setValueSerializer(new FastJsonRedisSerializer<>(Object.class));
                redisTemplate.setHashKeySerializer(new StringRedisSerializer());
                redisTemplate.setHashValueSerializer(new FastJsonRedisSerializer<>(Object.class));
                redisTemplate.afterPropertiesSet();
                return redisTemplate;
            }
            public Driverlogininfo TryGetData_DLD(String vehId, RedisTemplate<String, Object> clusterRedisTemplate) {
                Object obj = clusterRedisTemplate.opsForValue().get(RedisKeyPrefixConfig.RedisKey_DriverSignOn + vehId);
                if (obj == null) {
                    return new Driverlogininfo();
                } else {
                    return JSON.parseObject(JSON.toJSONString(obj, SerializerFeature.DisableCircularReferenceDetect), Driverlogininfo.class);
                }

            }

            public Tpvehdata TryGetData_Last(String simNo, RedisTemplate<String, Object> clusterRedisTemplate) {
                Object obj = clusterRedisTemplate.opsForValue().get(RedisKeyPrefixConfig.RedisKey_TpData + simNo);
                if (obj == null) {
                    return null;
                } else {
                    return JSON.parseObject(JSON.toJSONString(obj, SerializerFeature.DisableCircularReferenceDetect), Tpvehdata.class);
                }

            }

            @Override
            public void open(Configuration parameters) throws Exception {
                redisFactory = RedisSessionFactory.getRedisFactory();
            }

            @Override
            public VehGps map(GPSPosition gps) throws Exception {
                RedisTemplate<String, Object> clusterRedisTemplate = getRedisTemplate(redisFactory);
                Driverlogininfo driverlogininfo=TryGetData_DLD(gps.getVehId(),clusterRedisTemplate);
                VehGps vehGps = new VehGps();
                vehGps.setId(UUID.randomUUID().toString().replaceAll("-", ""));
                vehGps.setVehicleId(gps.getVehId());
                vehGps.setDeptId(gps.getDeptId());
                vehGps.setDriverName(driverlogininfo.getDriverName());
                vehGps.setDriverNo(driverlogininfo.getDriverNo());
                vehGps.setCheliangpaizhao(gps.getCheliangpaizhao());
                vehGps.setChepaiyanse(gps.getChepaiyanse());
                vehGps.setGpstime(gps.getGPSTimeExt());
                GeometryFactory factory = new GeometryFactory();
                vehGps.setGemo(factory.createPoint(new Coordinate(gps.getLngExt(), gps.getLatExt(), gps.getALTITUDE())));
                vehGps.setLat(gps.getLatExt());
                vehGps.setLng(gps.getLngExt());
                vehGps.setVelocity(gps.getSpeedExt());
                vehGps.setAngle(gps.getDIRECTION());
                vehGps.setAlarm(gps.getAlarmFlagVal() == 0 ? 0 : 1);
                vehGps.setLocate(GpsIntUtils.LocalExt(gps.getSTATUSBIT()));
                vehGps.setSystime(new Date());
                vehGps.setAlarmnote(gps.getAlarmNoteExt());
                vehGps.setRoadname("未知道路");
                vehGps.setShiyongxingzhi(gps.getShiyongxingzhi());

                Tpvehdata lasPost= TryGetData_Last(gps.getTerminalPhoneNo(),clusterRedisTemplate);


                if(lasPost!=null){
                    if(lasPost.getAcc()!=null&&lasPost.getAcc()==1){

                       double dist= PointUtils.getDistance(vehGps.getLat(),vehGps.getLng(),lasPost.getLatitude(),lasPost.getLongitude());
                       vehGps.setDistFromPrePoint(dist);
                    }else{
                        vehGps.setDistFromPrePoint(0d);
                    }
                    vehGps.setDistSum(lasPost.getMileToday());
                    vehGps.setDistTotal(lasPost.getMileTotal());
                }else{
                    vehGps.setDistFromPrePoint(0d);
                }


                if (StringUtils.isNotEmpty(vehGps.getRoadname())) {
                    vehGps.setRoadlevel(RoadUtils.GetRoadLevelByRoadName(vehGps.getRoadname()));
                } else {
                    vehGps.setRoadlevel(null);
                }

                vehGps.setAcc(GpsIntUtils.AccExt(gps.getSTATUSBIT()));
                if (CollectionUtil.isNotEmpty(gps.getGnssDataAdditionalInfo())) {
                    for (int i = 0; i < gps.getGnssDataAdditionalInfo().size(); i++) {
                        JTT808_AddtionalInfo addtionalInfo = gps.getGnssDataAdditionalInfo().get(i);

                        if (addtionalInfo.getID() == 0x30) {
                            JTT808_0x0200_0x30 jtt808_0x0200_0x30 = (JTT808_0x0200_0x30) addtionalInfo;
                            vehGps.setSignal(Byte.toUnsignedInt(jtt808_0x0200_0x30.getWiFiSignalStrength()));
                        }
                        if (addtionalInfo.getID() == 0x31) {
                            JTT808_0x0200_0x31 jtt808_0x0200_0x31 = (JTT808_0x0200_0x31) addtionalInfo;
                            vehGps.setSatellite(Byte.toUnsignedInt(jtt808_0x0200_0x31.getGNSSCount()));
                        }
                    }
                } else {
                    vehGps.setSignal(0);
                    vehGps.setSatellite(0);
                }

                vehGps.setMileage(gps.MileageExt());
                vehGps.setElevation(gps.getALTITUDE());

                return vehGps;
            }
        });

        DataStream<MonDaoludengjiSet> daoludengjiSetDataStream = executionEnvironment.addSource(new MonDaoludengjiSetSource());
        MapStateDescriptor<String, MonDaoludengjiSet> monDaoludengjiSetMapStateDescriptor = new MapStateDescriptor<String, MonDaoludengjiSet>("DaoludengjiSetMapState", String.class, MonDaoludengjiSet.class);
        BroadcastStream<MonDaoludengjiSet> daoludengjiSetBroadcastStream = daoludengjiSetDataStream.broadcast(monDaoludengjiSetMapStateDescriptor);

        SingleOutputStreamOperator<VehGps> vehGpsSingleOutputStreamOperator = gpsSingleOutputStreamOperator
                .connect(daoludengjiSetBroadcastStream)
                .process(new DaoludengjiBroadcastFun(monDaoludengjiSetMapStateDescriptor));


        //todo: GPS定位入库vehGpsSingleOutputStreamOperator
        vehGpsSingleOutputStreamOperator.addSink(new MybatisGPSSink());

        ZdaqFlow zdaqFlow = new ZdaqFlow(gpsSingleStream);
        zdaqFlow.processZdaq();

        BeidouZongDuanAlarmFlow beidouAlarmFlow = new BeidouZongDuanAlarmFlow(gpsSingleStream);
        DataStream<BaobiaoAlarmsummary> zongduanAlarmResultStream = beidouAlarmFlow.process();

        DataStream<Driverlogininfo> driverlogininfoDataStream = executionEnvironment.addSource(new DriverLoginInfoSource());
        MapStateDescriptor<String, Driverlogininfo> driverlogininfoMapStateDescriptor = new MapStateDescriptor<>("driverloginState", String.class, Driverlogininfo.class);
        BroadcastStream<Driverlogininfo> driverlogininfoBroadcastStream = driverlogininfoDataStream.broadcast(driverlogininfoMapStateDescriptor);
        SingleOutputStreamOperator<TCmd020064> cmd020064SingleOutputStreamOperator = gpsSingleStream.filter((gps) -> {
                    return StringUtils.isNotEmpty(gps.getJtt808_0x64_platformId())
                            || StringUtils.isNotEmpty(gps.getJtt808_0x67_platformId())
                            || StringUtils.isNotEmpty(gps.getJtt808_0x70_platformId());
                }).connect(driverlogininfoBroadcastStream)
                .process(new TCmd020064Process(driverlogininfoMapStateDescriptor));


        // todo: tcmd64入库
        cmd020064SingleOutputStreamOperator.addSink(new TCmd020064Sink());
        cmd020064SingleOutputStreamOperator
                .filter(d -> {
                    return d.getFlagState().equalsIgnoreCase("报警开始");
                })
                .map(new MapFunction<TCmd020064, BaobiaoAlarmsummaryRecord>() {
                    @Override
                    public BaobiaoAlarmsummaryRecord map(TCmd020064 tCmd020064) throws Exception {
                        int AlarmLevel = 0;
                        switch (tCmd020064.getAlarmlevel()) {
                            case "一级报警":
                                AlarmLevel = 1;
                                break;
                            case "二级报警":
                                AlarmLevel = 2;
                                break;
                        }
                        BaobiaoAlarmsummaryRecord baobiaoAlarmsummaryRecord = new BaobiaoAlarmsummaryRecord();
                        baobiaoAlarmsummaryRecord.setVehid(tCmd020064.getVehId());
                        baobiaoAlarmsummaryRecord.setBegintime(tCmd020064.getGpsTime());
                        baobiaoAlarmsummaryRecord.setAlarmtype(tCmd020064.getAlarmType());
                        baobiaoAlarmsummaryRecord.setLongitude(BigDecimal.valueOf(tCmd020064.getLon()));
                        baobiaoAlarmsummaryRecord.setLatitude(BigDecimal.valueOf(tCmd020064.getLat()));
                        baobiaoAlarmsummaryRecord.setVelocity(tCmd020064.getVelocity());
                        baobiaoAlarmsummaryRecord.setAngle(0);
                        baobiaoAlarmsummaryRecord.setRoadLevel("0");
                        baobiaoAlarmsummaryRecord.setRoadLimited(0);
                        baobiaoAlarmsummaryRecord.setKeeptime(0);
                        baobiaoAlarmsummaryRecord.setLevel(AlarmLevel);
                        baobiaoAlarmsummaryRecord.setAlarmreportid(tCmd020064.getAlarmNumber());
                        return baobiaoAlarmsummaryRecord;
                    }
                }).addSink(new BaobiaoAlarmsummaryRecordSink());

        executionEnvironment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);


        Pattern<TCmd020064, TCmd020064> pattern = Pattern.<TCmd020064>begin("first").where(new IterativeCondition<TCmd020064>() {
            @Override
            public boolean filter(TCmd020064 cmd020064, Context<TCmd020064> context) throws Exception {

                return cmd020064.getVelocity() > 60;
            }
        }).followedBy("second").where(new IterativeCondition<TCmd020064>() {
            @Override
            public boolean filter(TCmd020064 cmd020064, Context<TCmd020064> context) throws Exception {
                return cmd020064.getVelocity() > 60;
            }
        }).followedBy("third").where(new IterativeCondition<TCmd020064>() {
            @Override
            public boolean filter(TCmd020064 cmd020064, Context<TCmd020064> context) throws Exception {
                return cmd020064.getVelocity() > 60;
            }
        }).within(Time.minutes(5));


        PatternStream<TCmd020064> ps = CEP.pattern(cmd020064SingleOutputStreamOperator.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<TCmd020064>(Time.seconds(10)) {
            @Override
            public long extractTimestamp(TCmd020064 gpsPosition) {
                return gpsPosition.getGpsTime().getTime();
            }
        }).keyBy(gps -> gps.getVehId() + "_" + gps.getAlarmType()), pattern);

        SingleOutputStreamOperator<Riskinfo> riskinfoStreamToDatabase = ps.select(new PatternSelectFunction<TCmd020064, Riskinfo>() {
            @Override
            public Riskinfo select(Map<String, List<TCmd020064>> map) throws Exception {

                TCmd020064 first = map.get("first").iterator().next();
                TCmd020064 second = map.get("second").iterator().next();
                TCmd020064 third = map.get("third").iterator().next();

                Riskinfo riskinfo = new Riskinfo();
                riskinfo.setVehicleId(first.getVehId());
                riskinfo.setLocalTime(new Date());
                riskinfo.setRiskTime(third.getATime());
                riskinfo.setRiskNote(third.getAlarmType());
                riskinfo.setRiskLevel(3);
                riskinfo.setRiskSpeed(third.getVelocity());
                riskinfo.setAlarmNumber(third.getAlarmNumber());

                return riskinfo;
            }
        });
        //todo: riskinfo入库


        SingleOutputStreamOperator<TCmd020065> cmd020065SingleOutputStreamOperator = gpsSingleStream.filter((gps) -> {
            return StringUtils.isNotEmpty(gps.getJtt808_0x65_platformId());
        }).flatMap(new TCmd020065Process());

        //todo: tcmd65入库

        cmd020065SingleOutputStreamOperator.addSink(new TCmd020065Sink());

        Pattern<TCmd020065, TCmd020065> cmd020065Pattern = Pattern.<TCmd020065>begin("first").where(new IterativeCondition<TCmd020065>() {
            @Override
            public boolean filter(TCmd020065 cmd020064, Context<TCmd020065> context) throws Exception {

                return cmd020064.getVelocity() > 60;
            }
        }).followedBy("second").where(new IterativeCondition<TCmd020065>() {
            @Override
            public boolean filter(TCmd020065 cmd020064, Context<TCmd020065> context) throws Exception {
                return cmd020064.getVelocity() > 60;
            }
        }).followedBy("third").where(new IterativeCondition<TCmd020065>() {
            @Override
            public boolean filter(TCmd020065 cmd020064, Context<TCmd020065> context) throws Exception {
                return cmd020064.getVelocity() > 60;
            }
        }).within(Time.minutes(5));

        PatternStream<TCmd020065> ps65 = CEP.pattern(cmd020065SingleOutputStreamOperator.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<TCmd020065>(Time.seconds(10)) {
            @Override
            public long extractTimestamp(TCmd020065 gpsPosition) {
                return gpsPosition.getGpsTime().getTime();
            }
        }).keyBy(gps -> gps.getVehId() + "_" + gps.getAlarmType()), cmd020065Pattern);

        SingleOutputStreamOperator<Riskinfo> riskinfo65Stream = ps65.select(new PatternSelectFunction<TCmd020065, Riskinfo>() {
            @Override
            public Riskinfo select(Map<String, List<TCmd020065>> map) throws Exception {

                TCmd020065 first = map.get("first").iterator().next();
                TCmd020065 second = map.get("second").iterator().next();
                TCmd020065 third = map.get("third").iterator().next();
                Riskinfo riskinfo = new Riskinfo();
                riskinfo.setVehicleId(first.getVehId());
                riskinfo.setLocalTime(new Date());
                riskinfo.setRiskTime(third.getATime());
                riskinfo.setRiskNote(third.getAlarmType());
                riskinfo.setRiskLevel(3);
                riskinfo.setRiskSpeed(third.getVelocity());
                riskinfo.setAlarmNumber(third.getAlarmNumber());

                return riskinfo;
            }
        });

        //todo: riskinfo65Stream入库
        riskinfo65Stream.print();

        // 分段限速
        RoadLevelProcess roadLevelProcess = new RoadLevelProcess(gpsSingleStream);
        roadLevelProcess.process();

        // 疲劳驾驶 连续驾驶4个小时休息20分钟

        FatigueDrivingAlarmFlow fatigueDrivingAlarmFlow = new FatigueDrivingAlarmFlow(gpsSingleStream);
        fatigueDrivingAlarmFlow.process();


        //累计疲劳驾驶

        TotalFatigueDrivingFlow totalFatigueDrivingFlow = new TotalFatigueDrivingFlow(gpsSingleStream);
        DataStream<BaobiaoAlarmsummary> totalFatigueAlarmResult = totalFatigueDrivingFlow.process();


        //夜间禁行
        ProhibitDrivingFlow prohibitDrivingFlow = new ProhibitDrivingFlow(gpsSingleStream);
        DataStream<BaobiaoAlarmsummary> prohibitDrivingAlarmResult = prohibitDrivingFlow.process();

        //无数据
        NoPositionFlow noPositionFlow = new NoPositionFlow(gpsSingleStream);
        DataStream<BaobiaoAlarmsummary> noPositionAlarmResult = noPositionFlow.process();

        //不定位
        NoLocateFlow noLocateFlow = new NoLocateFlow(gpsSingleStream);
        DataStream<BaobiaoAlarmsummary> noLocateAlarmResult = noLocateFlow.process();
        //异常掉线
        AbNormalofflineFlow abNormalofflineFlow=new AbNormalofflineFlow(gpsSingleStream);
        DataStream<BaobiaoAlarmsummary> abNormalofflineAlarmResult = abNormalofflineFlow.process();
        //进出区域
        AreaAlarmFlow areaAlarmFlow=new AreaAlarmFlow(gpsSingleStream);
        areaAlarmFlow.process();
        //风险区
        AreaRiskAlarmFlow areaRiskAlarmFlow=new AreaRiskAlarmFlow(gpsSingleStream);
        areaRiskAlarmFlow.process();
        //最新位置更新
        UpdateTpDataFlow updateTpDataFlow=new UpdateTpDataFlow(gpsSingleStream);
        updateTpDataFlow.process();





        Time windowSize = Time.minutes(1);
        executionEnvironment.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);


        SingleOutputStreamOperator<VehGps> gpsWater = vehGpsSingleOutputStreamOperator.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<VehGps>(windowSize) {
            @Override
            public long extractTimestamp(VehGps gpsPosition) {
                return gpsPosition.getSystime().getTime();
            }
        });


        SingleOutputStreamOperator<BaobiaoAlarmsummary> zongduanWater = zongduanAlarmResultStream.filter(e -> e.getBegintime() != null && e.getEndtime() == null)
                .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<BaobiaoAlarmsummary>(windowSize) {
                    @Override
                    public long extractTimestamp(BaobiaoAlarmsummary baobiaoAlarmsummary) {
                        return baobiaoAlarmsummary.getSyntime().getTime();
                    }
                });

        SingleOutputStreamOperator<BaobiaoAlarmsummary> totalFatigueWater = totalFatigueAlarmResult.filter(e -> e.getBegintime() != null && e.getEndtime() == null)
                .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<BaobiaoAlarmsummary>(windowSize) {
                    @Override
                    public long extractTimestamp(BaobiaoAlarmsummary baobiaoAlarmsummary) {
                        return baobiaoAlarmsummary.getSyntime().getTime();
                    }
                });

        gpsWater.coGroup(zongduanWater).where(e -> e.getVehicleId()).equalTo(e -> e.getVehid()).window(TumblingProcessingTimeWindows.of(windowSize))
                .apply(new CoGroupFunction<VehGps, BaobiaoAlarmsummary, Tuple3<String, List<GPSPosition>, List<BaobiaoAlarmsummary>>>() {
                    @Override
                    public void coGroup(Iterable<VehGps> iterable, Iterable<BaobiaoAlarmsummary> iterable1, Collector<Tuple3<String, List<GPSPosition>, List<BaobiaoAlarmsummary>>> collector) throws Exception {

                    }
                }).print();


        executionEnvironment.execute("gzyj-daoluyunshu-gps");
    }
}
